package com.atguigu.utils;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * kafka自定义Consumer拉取数据
 *
 * @author WangJX
 * @version 1.0
 * @date 2019/9/27 15:45
 */
public class CustomConsumerAutoOffset {
    public static void main(String[] args) throws IOException {
        Properties prop = new Properties();
        prop.load(CustomConsumerAutoOffset.class.getClassLoader().getResourceAsStream("kafkaconsumer.properties"));

        KafkaConsumer<String, String> kafkaConsumer = null;
        try {
            kafkaConsumer = new KafkaConsumer<>(prop);

            //设置一个单集合的topic主题
            kafkaConsumer.subscribe(Collections.singleton("GMALL_NEW_ORDER"));
//            kafkaConsumer.subscribe(Collections.singleton("second"));

            //循环拉取数据
            while (true) {
                //设置多少秒拉取一次数据
                ConsumerRecords<String, String> poll = kafkaConsumer.poll(1000);
                //ConsumerRecords<String, String> poll = kafkaConsumer.poll(Duration.ofSeconds(100));

                for (ConsumerRecord<String, String> consumerRecord : poll) {
                    System.out.println(consumerRecord);
                }
            }
        } finally {
            kafkaConsumer.close();
        }

    }
}
